/**
 * 
 */
package com.googlecode.xcarve.transport;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.googlecode.xcarve.coder.NettyCodecAdapter;
import com.googlecode.xcarve.common.config.CarveConstances;
import com.googlecode.xcarve.config.ConfigRefer;
import com.googlecode.xcarve.netty.handler.NettyClientHandler;
import com.googlecode.xcarve.thread.DoveThreadFactory;
import com.googlecode.xcarve.util.ConfigHolder;

/**
 * @author duxinyun
 *
 */
public abstract class AbstractClient implements Client{

	private  Bootstrap              bootStrap;
	public   final ConfigRefer     config;
	protected volatile Channel	    channel;
	protected volatile boolean	    closed = false;
	public static final Map<String,Client> clientMap = new HashMap<String,Client>();
	
	private static final EventLoopGroup bossGroup = new NioEventLoopGroup(0, 
			Executors.newCachedThreadPool(new DoveThreadFactory("dove client boss group")));
	
	private static final ScheduledThreadPoolExecutor scheduledService = new ScheduledThreadPoolExecutor(2, new DoveThreadFactory("DoveReconnect"));
	
	private final Lock lock = new ReentrantLock();
	private volatile  ScheduledFuture<?> reconnectExecutorFuture = null;
	private static final Logger logger = LoggerFactory.getLogger(AbstractClient.class);
	
	
	public AbstractClient(ConfigRefer config) {
		this.config = config;
		this.bootStrap = createBootStrap();
		connection();
	}
	
	Bootstrap createBootStrap() {
		Bootstrap bootStrap = new Bootstrap();
		final NettyClientHandler handler = new NettyClientHandler();
		bootStrap.group(bossGroup).channel(NioSocketChannel.class)
		.option(ChannelOption.TCP_NODELAY, true)
		.option(ChannelOption.SO_KEEPALIVE, true)
		.handler(new ChannelInitializer<SocketChannel>() {

			@Override
			protected void initChannel(SocketChannel ch) throws Exception {
				ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
				NettyCodecAdapter coder = new NettyCodecAdapter(config.getCoder());
				ch.pipeline().addLast("encoder",coder.getEncoderHandler());
				ch.pipeline().addLast("decoder",coder.getDecoderHandler());
				ch.pipeline().addLast(handler);
			}
			
		});
		return bootStrap;
	}
	
	protected void connection() throws RuntimeException{
		try {
			lock.lock();
			if( !isClosed() && isConnection()){
				logger.info("this connection is valid, so do nothing. address = {}", config.getAddress());
				return;
			}
			logger.info("xcarve start to create remote connection address = {}", config.getAddress());
			initSchedual();
			doConnection();
		} catch (Exception e) {
			logger.error("xcarve create remote connection error.",e);
		}finally{
			lock.unlock();
		}
	}
	
	public synchronized void initSchedual(){
		 int idel = config.getIdleTimeMillis();
		 if(idel <= 0){
		 	 idel = CarveConstances.DEFAULT_IDLETIME_PERIOD;
		 }
		 ConfigHolder.getIntValue("", CarveConstances.DEFAULT_IDLETIME_PERIOD);
		 if(reconnectExecutorFuture != null && !reconnectExecutorFuture.isCancelled()){
			 return ;
		 }
		 Runnable command = new Runnable() {
			
			@Override
			public void run() {
				try {
					if(!isConnection() || isClosed()){
						logger.info("old connect is not available. create a new connect right now!");
						connection();
					}
				} catch (Exception e) {
					logger.error("connect to server error.",e);
				}
			}
		};
		reconnectExecutorFuture = scheduledService.scheduleWithFixedDelay(command, idel, idel, TimeUnit.MILLISECONDS);
	}
	
	
	public abstract void doConnection() throws RuntimeException;
	
	public boolean isConnection(){
		if(channel == null || !channel.isOpen()){
			return false;
		}
		return channel.isOpen();
	}
	
	@Override
	public Channel getConnection() {
		Channel ch = channel;
		if(ch == null || !ch.isOpen()){
			close();
			return null;
		}
		return ch;
	}
	
	public void reconnection(){
		lock.lock();
		try {
			if( reconnectExecutorFuture !=null || !reconnectExecutorFuture.isDone()){
				reconnectExecutorFuture.cancel(true);
				scheduledService.purge();
				logger.info("xcarve cancel net reconnection executors ");
			}
			close();
			connection();
		} catch (Exception e) {
			logger.error("xcarve reconnect to server error, server_address="+config.getAddress(), e);
		}finally{
			lock.unlock();
		}
	}

	@Override
	public void send(final Object msg)throws Exception{
		send(msg,CarveConstances.DEFAULT_TIMEOUT);
	}
	
	public void send(Object msg,final long timeout)throws Exception {
//		if(!isConnection()){
//			connection();
//		}
		if(getConnection() == null || !channel.isOpen()){
			throw new RuntimeException("message can not send, because channel is closed .");
		}
		doSend(msg,timeout);
	}

	public void close(){
		closed = true;
		clientMap.remove(config.getAddress());
		if( reconnectExecutorFuture !=null || !reconnectExecutorFuture.isDone()){
			reconnectExecutorFuture.cancel(true);
			scheduledService.purge();
		}
		if(scheduledService != null || !scheduledService.isShutdown()){
			scheduledService.shutdown();
		}
		logger.info("xcarve close net reconnection scheduledService ");
		if(channel !=null || channel.isOpen()){
			channel.close();
		}
	}
	
	public boolean isClosed(){
		return closed;
	}

	public void connected(){
		closed = false;
	}
	
	public abstract void doSend(final Object msg,final long timeout) throws Exception;
	
	/**
	 * @return the bootStrap
	 */
	public Bootstrap getBootStrap() {
		return bootStrap;
	}
}
